package com.atguigu.flink.edu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.flink.edu.app.fun.DimAsyncFunction;
import com.atguigu.flink.edu.beans.TradeProvinceOrderBean;
import com.atguigu.flink.edu.utils.DataFormatUtil;
import com.atguigu.flink.edu.utils.MyClickhouseUtil;
import com.atguigu.flink.edu.utils.MyKafkaUtil;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.sql.Connection;
import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

public class dws_Trade_Order_Province_Window {
    public static void main(String[] args) throws Exception {
        //TODO 1 :基本环境设置
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        env.disableOperatorChaining();
        //TODO 2 :获取kafka数据
        String topic = "dwd_trade_order_detail_inc";
        String groupId = "dwd_trade_order_detail_group";
        KafkaSource<String> sourceStr = MyKafkaUtil.getKafkaSource(topic, groupId);
        //2.1转换为流数据
        DataStreamSource<String> kafkaSourceStr = env.fromSource(sourceStr, WatermarkStrategy.noWatermarks(), "source_kafak");
        //2.2 类型转换
        SingleOutputStreamOperator<JSONObject> jsonObjDs = kafkaSourceStr.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String sourceStr, Context context, Collector<JSONObject> out) throws Exception {
                JSONObject jsonObj = JSON.parseObject(sourceStr);
                out.collect(jsonObj);
            }
        });
        //TODO 3:按照order_id进行分组
        KeyedStream<JSONObject, String> keyedDs = jsonObjDs.keyBy(r -> r.getString("order_id"));
        //TODO 4:去重
        SingleOutputStreamOperator<JSONObject> processDs = keyedDs.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            private ValueState<JSONObject> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                valueState = getRuntimeContext().getState(
                        new ValueStateDescriptor<JSONObject>(
                                "value_order_id",
                                JSONObject.class
                        )
                );
            }

            @Override
            public void processElement(JSONObject jsonObject, Context context, Collector<JSONObject> collector) throws Exception {
                JSONObject value = valueState.value();
                if (value != null) {
                    String final_amount = jsonObject.getString("final_amount");
                    value.put("final_amount", "-" + final_amount);
                    collector.collect(value);
                }
                valueState.update(jsonObject);
                collector.collect(jsonObject);
                //System.out.println("==>"+jsonObject.getLong("ts"));
            }
        });
        //TODO 4:转换类型
        SingleOutputStreamOperator<TradeProvinceOrderBean> orderBean = processDs.map(new MapFunction<JSONObject, TradeProvinceOrderBean>() {
            @Override
            public TradeProvinceOrderBean map(JSONObject jsonObject) throws Exception {
                String order_id = jsonObject.getString("order_id");
                TradeProvinceOrderBean provinceOrderBean = TradeProvinceOrderBean.builder()
                        .ts(jsonObject.getLong("ts") * 1000)
                        .orderAmount(new BigDecimal(jsonObject.getString("final_amount")))
                        .provinceId(jsonObject.getString("province_id"))
                        .orderIdSet(new HashSet(Collections.singleton(order_id)))
                        .build();
                return provinceOrderBean;
            }
        });
        System.out.println();
        //orderBean.print("----->");
        //TODO 5:设置水位线
        SingleOutputStreamOperator<TradeProvinceOrderBean> withWaterMarkDs = orderBean.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TradeProvinceOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner(new SerializableTimestampAssigner<TradeProvinceOrderBean>() {
                            @Override
                            public long extractTimestamp(TradeProvinceOrderBean orderbean, long l) {
                                return orderbean.getTs();
                            }
                        }));
        //TODO 6:按照省份ID进行分组
        KeyedStream<TradeProvinceOrderBean, String> keyedPidDs = withWaterMarkDs.keyBy(r -> r.getProvinceId());
        //TODO 7:开窗
        WindowedStream<TradeProvinceOrderBean, String, TimeWindow> windowDs
                = keyedPidDs.window(TumblingEventTimeWindows.of(Time.seconds(10)));
        //TODO 8:聚合
        SingleOutputStreamOperator<TradeProvinceOrderBean> reduceDs
                = windowDs.reduce(new ReduceFunction<TradeProvinceOrderBean>() {
                                      @Override
                                      public TradeProvinceOrderBean reduce(TradeProvinceOrderBean value1, TradeProvinceOrderBean value2) throws Exception {
                                          value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                                          value1.getOrderIdSet().add(value2.getOrderIdSet());
                                          return value1;
                                      }
                                  }, new WindowFunction<TradeProvinceOrderBean, TradeProvinceOrderBean, String, TimeWindow>() {
                                      @Override
                                      public void apply(String orderbean, TimeWindow timeWindow, Iterable<TradeProvinceOrderBean> iterable, Collector<TradeProvinceOrderBean> collector) throws Exception {
                                          for (TradeProvinceOrderBean bean : iterable) {
                                              bean.setStt(DataFormatUtil.toYmdHms(timeWindow.getStart()));
                                              bean.setEdt(DataFormatUtil.toYmdHms(timeWindow.getEnd()));
                                              bean.setOrderCount((long) bean.getOrderIdSet().size());
                                              bean.setTs(System.currentTimeMillis());
                                              collector.collect(bean);
                                              System.out.println(bean.getTs());
                                          }
                                      }
                                  }
        );
        //TODO 9:省份表维度关联
        //reduceDs.print("==>");
        SingleOutputStreamOperator withProDs = AsyncDataStream.unorderedWait(
                reduceDs,
                new DimAsyncFunction<TradeProvinceOrderBean>("dim_base_province") {
                    @Override
                    public void join(TradeProvinceOrderBean obj, JSONObject dimJsonObj) throws Exception {
                        obj.setProvinceName(dimJsonObj.getString("NAME"));
                    }
                    @Override
                    public String getKey(TradeProvinceOrderBean obj) {
                        return obj.getProvinceId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        withProDs.print();
        //TODO 10:将省份表写入CK
        //写入CK
        withProDs.addSink(
                MyClickhouseUtil.getSinkFunction("insert into dws_trade_province_order_window values(?,?,?,?,?,?,?)")
        );
        env.execute();
    }
}
